package com.zallds.window;

import com.zallds.utils.Utils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 增量聚合函数
 * ReduceFunction
 * 每经过3秒钟，求安放在北京西站各个红外测温仪所探测到的体温最高的旅客信息
 */
public class InrementAggregateFunctionDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8888)
                .flatMap(new Utils.Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceCount())
                .print();
        env.execute();

    }

    public static class ReduceCount implements ReduceFunction<Tuple2<String, Integer>> {

        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t0, Tuple2<String, Integer> t1) throws Exception {
            return t1.f1 > t0.f1 ? t1 : t0;
        }
    }
}
